package com.swt.testdemo.util;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.text.StrBuilder;
import cn.hutool.core.text.csv.CsvRow;
import cn.hutool.core.util.StrUtil;
import cn.hutool.crypto.SecureUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.swt.testdemo.exception.AgyException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * ES查询帮助类
 *
 * @author BleethNie
 * @version 1.0
 * @date 2020-03-18 17:15
 **/
@Component
@Slf4j
public class ESQueryUtil {

    private static final String ES_SETTING_NAME = "swt.elasticsearch.cluster.node";

    public static final String ES_INDEX_TYPE_SOURCE = "source";

    public static final String ES_INDEX_TYPE_NODE = "node";


    @Value("${swt.elasticsearch.cluster.node}")
    private String[] esNodes;



    String _idStr = "";

    public List<String> getIndexMapping(String indexName, String type) {
        String getUrl = StrUtil.format("http://{}/{}", esNodes[0], indexName);
        HttpRequest httpRequest = HttpRequest.get(getUrl);
        String resultJSON = httpRequest.header("Connection", "close")
                .keepAlive(false)
                .timeout(20000)//超时，毫秒
                .execute().body();
        boolean isJSON = JSONUtil.isJson(resultJSON);
        if (!isJSON) {
            throw new AgyException(StrUtil.format("ES查询异常,返回结果异常"));
        }
        JSONObject rootJSON = JSONUtil.parseObj(resultJSON);
        JSONObject errorJSON = rootJSON.getJSONObject("error");
        if (errorJSON != null) {
            return null;
        }
        JSONObject indexObject = rootJSON.getJSONObject(indexName);
        JSONObject mappingsObject = indexObject.getJSONObject("mappings");
        JSONObject sourceObject = mappingsObject.getJSONObject(type);
        JSONObject propertiesObject = sourceObject.getJSONObject("properties");
        List<String> headerList = new ArrayList<>();
        propertiesObject.entrySet().stream().forEach(entry -> {
            String key = entry.getKey();
            headerList.add(key);
        });

        return headerList;
    }

    public void deleteEsRow(String indexName, String type, List<String> _idList) {
        String metaTemplate = "{ \"delete\": { \"_index\": \"{}\", \"_type\": \"{}\", \"_id\": \"{}\" }}";
        StrBuilder strBuilder = new StrBuilder();
        for (int i = 0; i < _idList.size(); i++) {
            String _id = _idList.get(i);
            String meta = StrUtil.format(metaTemplate, indexName, type, _id);
            strBuilder.append(meta).append("\r\n");
        }
        String postUrl = StrUtil.format("http://{}/_bulk", esNodes[0]);
        HttpRequest httpRequest = HttpRequest.post(postUrl);
        String resultJSON = httpRequest.header("Connection", "close")
                .keepAlive(false)
                .body(strBuilder.toString())
                .timeout(20000)//超时，毫秒
                .execute().body();
        boolean isJSON = JSONUtil.isJson(resultJSON);
        if (!isJSON) {
            throw new AgyException(StrUtil.format("ES查询异常,返回结果异常"));
        }
    }


    public boolean checkIndex(String indexName) {
        String getUrl = StrUtil.format("http://{}/{}", esNodes[0], indexName);
        HttpRequest httpRequest = HttpRequest.get(getUrl);
        String resultJSON = httpRequest.header("Connection", "close")
                .keepAlive(false)
                .timeout(20000)//超时，毫秒
                .execute().body();
        boolean isJSON = JSONUtil.isJson(resultJSON);
        if (!isJSON) {
            throw new AgyException(StrUtil.format("ES查询异常,返回结果异常"));
        }
        JSONObject rootJSON = JSONUtil.parseObj(resultJSON);
        JSONObject errorJSON = rootJSON.getJSONObject("error");
        if (errorJSON == null) {
            return true;
        }
        return false;
    }


    public boolean createIndexAndMapping(String indexName, String type, List<String> header) {
        //创建mappping
        String putUrl = "http://" + esNodes[0] + "/" + indexName;

        JSONObject root = new JSONObject();
        JSONObject mappingObject = new JSONObject();
        JSONObject typeObject = new JSONObject();
        JSONObject propertiesObject = new JSONObject();
        for (int i = 0; i < header.size(); i++) {
            String field = header.get(i);
            String fieldType = "{\"analyzer\":\"hanlp_index\",\"type\":\"text\",\"fields\":{\"keyword\":{\"type\":\"keyword\"}}}";
            JSONObject fieldObj = JSONUtil.parseObj(fieldType);
            propertiesObject.set(field, fieldObj);
        }

        //添加系统字段
        String dateType = "{\"format\":\"yyyy-MM-dd HH:mm:ss\",\"type\":\"date\"}";
        JSONObject fieldObj = JSONUtil.parseObj(dateType);
        propertiesObject.set("_updateTime", fieldObj);


        typeObject.set("properties", propertiesObject);
        mappingObject.set(type, typeObject);
        root.set("mappings", mappingObject);


        String resultJSON = HttpRequest
                .put(putUrl)
                .header("Connection", "close")
                .keepAlive(false)
                .body(root.toStringPretty())
                .timeout(20000)//超时，毫秒
                .execute().body();
        boolean isJSON = JSONUtil.isJson(resultJSON);
        if (!isJSON) {
            throw new AgyException(StrUtil.format("ES查询异常,返回结果异常"));
        }
        JSONObject rootJSON = JSONUtil.parseObj(resultJSON);
        JSONObject errorJSON = rootJSON.getJSONObject("error");
        if (errorJSON == null) {
            return false;
        }
        return true;
    }

    public boolean commitSource(String indexName, String type, List<String> header, List<CsvRow> csvRowList) {
        //创建mappping
        String postUrl = StrUtil.format("http://{}/_bulk", esNodes[0]);
        String metaTemplate = "{\"index\": {\"_index\": \"{}\", \"_type\": \"{}\", \"_id\": \"{}\"}}";
        String docTemplate = "{}";
        StrBuilder strBuilder = new StrBuilder();

        DateTime current = DateUtil.date();

        for (CsvRow csvRow : csvRowList) {
            String _idStr = "";
            JSONObject docObject = new JSONObject();
            JSONObject json = new JSONObject();
            for (int i = 0; i < header.size(); i++) {
                String field = header.get(i);
                String value = csvRow.getByName(field);
                _idStr = _idStr + value;
                docObject.set(field, value);
            }

            //系统字段
            docObject.set("_updateTime", current.toString());


            String _id = SecureUtil.md5(_idStr);
            String meta = StrUtil.format(metaTemplate, indexName, type, _id);
            String doc = StrUtil.format(docTemplate, docObject.toString());
            strBuilder.append(meta).append("\r\n").append(doc).append("\r\n");
        }

        String resultJSON = HttpRequest.post(postUrl)
                .header("Connection", "close")
                .keepAlive(false)
                .body(strBuilder.toString())
                .timeout(20000)//超时，毫秒
                .execute().body();
        boolean isJSON = JSONUtil.isJson(resultJSON);
        if (!isJSON) {
            throw new AgyException(StrUtil.format("ES查询异常,返回结果异常"));
        }
        return true;
    }


    public boolean createIndex(String indexName) {
        String putUrl = StrUtil.format("http://{}/{}", esNodes[0], indexName);
        String resultJSON = HttpRequest.put(putUrl)
                .header("Connection", "close")
                .keepAlive(false)
                .timeout(20000)//超时，毫秒
                .execute().body();
        boolean isJSON = JSONUtil.isJson(resultJSON);
        if (!isJSON) {
            throw new AgyException(StrUtil.format("ES查询异常,返回结果异常"));
        }
        JSONObject rootJSON = JSONUtil.parseObj(resultJSON);
        JSONObject errorJSON = rootJSON.getJSONObject("error");
        if (errorJSON == null) {
            return false;
        }
        return true;
    }




    /**
     * @param response
     * @return
     */
    public Long parseHitsCount(JSONObject response) {
        JSONObject hitsObject = response.getJSONObject("hits");
        return hitsObject.getLong("total");
    }


    /**
     * @param response
     * @param clazz
     * @param <T>
     * @return
     */
    public <T> List<T> parseHit(JSONObject response, Class<T> clazz) {
        JSONObject hitsObject = response.getJSONObject("hits");
        JSONArray hitsArray = hitsObject.getJSONArray("hits");
        if (hitsArray == null || hitsArray.size() == 0) {
            return new ArrayList<>();
        }
        List<T> ts = new ArrayList<>();
        for (int i = 0; i < hitsArray.size(); i++) {
            JSONObject hitObject = hitsArray.getJSONObject(i);
            ts.add(hitObject.getBean("_source", clazz));
        }
        return ts;
    }

    /**
     * @param response
     * @return
     */
    public List<JSONObject> parseHit(JSONObject response) {
        JSONObject hitsObject = response.getJSONObject("hits");
        JSONArray hitsArray = hitsObject.getJSONArray("hits");
        if (hitsArray == null || hitsArray.size() == 0) {
            return new ArrayList<>();
        }
        List<JSONObject> ts = new ArrayList<>();
        for (int i = 0; i < hitsArray.size(); i++) {
            JSONObject hitObject = hitsArray.getJSONObject(i);
            JSONObject sourceObject = hitObject.getJSONObject("_source");
            String _id = hitObject.getStr("_id");
            sourceObject.set("esId", _id);
            ts.add(sourceObject);
        }
        return ts;
    }


    /**
     * @param response
     * @param aggName
     * @return
     */
    public JSONArray parseBuckets(JSONObject response, String aggName) {
        JSONObject agg;
        JSONObject aggObject = response.getJSONObject("aggs");
        JSONObject aggregationsObject = response.getJSONObject("aggregations");
        if (aggObject == null && aggregationsObject == null) {
            log.info("结果参数：" + response.toStringPretty());
            throw new AgyException(StrUtil.format("ES查询异常,聚合结果异常"));
        }
        agg = aggObject != null ? aggObject : aggregationsObject;
        JSONObject aggResultObject = agg.getJSONObject(aggName);
        return aggResultObject.getJSONArray("buckets");
    }


    public Integer parseDistinct(JSONObject response) {
        JSONObject agg;
        JSONObject aggObject = response.getJSONObject("aggs");
        JSONObject aggregationsObject = response.getJSONObject("aggregations");
        if (aggObject == null && aggregationsObject == null) {
            log.info("结果参数：" + response.toStringPretty());
            throw new AgyException(StrUtil.format("ES查询异常,聚合结果异常"));
        }
        agg = aggObject != null ? aggObject : aggregationsObject;
        JSONObject distinctObject = agg.getJSONObject("distinct");
        return distinctObject.getInt("value");
    }


    public boolean removeIndex(String indexName) {
        //创建mappping
        String deleteUrl = "http://" + esNodes[0] + "/" + indexName;

        String resultJSON = HttpRequest
                .delete(deleteUrl)
                .header("Connection", "close")
                .keepAlive(false)
                .timeout(20000)//超时，毫秒
                .execute().body();
        boolean isJSON = JSONUtil.isJson(resultJSON);
        if (!isJSON) {
            throw new AgyException(StrUtil.format("ES查询异常,返回结果异常"));
        }
        JSONObject rootJSON = JSONUtil.parseObj(resultJSON);
        JSONObject errorJSON = rootJSON.getJSONObject("error");
        if (errorJSON == null) {
            return false;
        }
        return true;

    }

    public String scrollSearch(String sourceIndex, String scrollId) {
        if (StrUtil.isEmpty(scrollId)) {
            String getUrl = StrUtil.format("http://{}/{}/_search?scroll=2m", esNodes[0], sourceIndex);
            String queryTemplate = "{\"query\":{\"match_all\":{}},\"sort\":[\"_doc\"],\"size\":1000}";

            String response = HttpRequest.get(getUrl)
                    .header("Connection", "close")
                    .body(queryTemplate)
                    .keepAlive(false)
                    .timeout(20000)
                    .execute()
                    .body();
            return response;
        }

        String getUrl = StrUtil.format("http://{}/{}/_search/scroll", esNodes[0], sourceIndex);
        String queryTemplate = "{\"scroll\":\"2m\",\"scroll_id\":\"{}\"}";
        String query = StrUtil.format(queryTemplate, scrollId);
        String response = HttpRequest.get(getUrl)
                .header("Connection", "close")
                .body(query)
                .keepAlive(false)
                .timeout(20000)
                .execute()
                .body();
        return response;
    }


    public String insertNodeData(Integer nodeId, String realEsId, JSONObject importNodeObject) {
        String index = "node_" + nodeId;
        String putUrl = StrUtil.format("http://{}/{}/{}/{}", esNodes[0], index, "node", realEsId);

        String response = HttpRequest.put(putUrl)
                .header("Connection", "close")
                .body(importNodeObject.toStringPretty())
                .keepAlive(false)
                .timeout(20000)
                .execute()
                .body();
        return response;
    }


    public boolean createSourceIndexAndMapping(String sourceHeader, String index) {
        //创建mappping
        String putUrl = "http://" + esNodes[0] + "/" + index;


        List<String> headerList = JSONUtil.toList(JSONUtil.parseArray(sourceHeader), String.class);


        JSONObject root = new JSONObject();
        JSONObject mappingObject = new JSONObject();
        JSONObject typeObject = new JSONObject();
        JSONObject propertiesObject = new JSONObject();
        for (int i = 0; i < headerList.size(); i++) {
            String field = headerList.get(i);
            String fieldType = "{\"analyzer\":\"hanlp_index\",\"type\":\"text\",\"fields\":{\"keyword\":{\"type\":\"keyword\"}}}";
            JSONObject fieldObj = JSONUtil.parseObj(fieldType);
            propertiesObject.set(field, fieldObj);
        }

        //添加系统字段
        String dateType = "{\"format\":\"yyyy-MM-dd HH:mm:ss\",\"type\":\"date\"}";
        JSONObject fieldObj = JSONUtil.parseObj(dateType);
        propertiesObject.set("_updateTime", fieldObj);

        dateType = "{\"type\":\"keyword\"}";
        fieldObj = JSONUtil.parseObj(dateType);
        propertiesObject.set("_status", fieldObj);


        typeObject.set("properties", propertiesObject);
        mappingObject.set(ES_INDEX_TYPE_SOURCE, typeObject);
        root.set("mappings", mappingObject);


        String resultJSON = HttpRequest
                .put(putUrl)
                .header("Connection", "close")
                .keepAlive(false)
                .body(root.toStringPretty())
                .timeout(20000)//超时，毫秒
                .execute().body();
        boolean isJSON = JSONUtil.isJson(resultJSON);
        if (!isJSON) {
            throw new AgyException(StrUtil.format("ES查询异常,返回结果异常"));
        }

        JSONObject rootJSON = JSONUtil.parseObj(resultJSON);
        JSONObject errorJSON = rootJSON.getJSONObject("error");
        if (errorJSON == null) {
            return false;
        }
        return true;
    }




    public void commitSource(Integer sourceId, List<Map<String, String>> rowList) {
        String indexName = ES_INDEX_TYPE_SOURCE + "_" + sourceId;
        String type = ES_INDEX_TYPE_SOURCE;
        //创建mappping
        String postUrl = StrUtil.format("http://{}/_bulk", esNodes[0]);
        String metaTemplate = "{\"index\": {\"_index\": \"{}\", \"_type\": \"{}\", \"_id\": \"{}\"}}";
        String docTemplate = "{}";
        StrBuilder strBuilder = new StrBuilder();

        DateTime current = DateUtil.date();

        for (Map<String, String> map : rowList) {
            _idStr = "";
            JSONObject docObject = new JSONObject();
            map.entrySet().stream().forEach(entry -> {
                String key = entry.getKey();
                String value = entry.getValue();
                _idStr = _idStr + value;
                docObject.set(key, value);
            });


            //系统字段
            docObject.set("_updateTime", current.toString());
            docObject.set("_status", "0");

            String _id = SecureUtil.md5(_idStr);
            String meta = StrUtil.format(metaTemplate, indexName, type, _id);
            String doc = StrUtil.format(docTemplate, docObject.toString());
            strBuilder.append(meta).append("\r\n").append(doc).append("\r\n");
        }

        String resultJSON = HttpRequest.post(postUrl)
                .header("Connection", "close")
                .keepAlive(false)
                .body(strBuilder.toString())
                .timeout(20000)//超时，毫秒
                .execute().body();
        boolean isJSON = JSONUtil.isJson(resultJSON);
        if (!isJSON) {
            throw new AgyException(StrUtil.format("ES查询异常,返回结果异常"));
        }
    }


    public JSONObject queryNode(String nodeIndex, String nodeId) {
        String getUrl = StrUtil.format("http://{}/{}/{}/{}", esNodes[0], nodeIndex, ES_INDEX_TYPE_NODE, nodeId);

        String response = HttpRequest.get(getUrl)
                .header("Connection", "close")
                .keepAlive(false)
                .timeout(20000)
                .execute()
                .body();

        JSONObject jsonObject = JSONUtil.parseObj(response);
        List<JSONObject> hitList = parseHit(jsonObject);
        if (hitList == null || hitList.isEmpty()) {
            throw new AgyException("未找到对应的节点信息");
        }
        return hitList.get(0);
    }


}
